package cn.itcast.flink.chk;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Author itcast
 * Date 2022/1/16 16:17
 * Desc 每 5s 实现将状态保存到 HDFS 上，wordcount
 */
public class CheckpointDemo {
    public static void main(String[] args) throws Exception {
        // 实现步骤：
        // 1）初始化flink流处理的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 2）开启checkpoint ，存储到内存中
        env.enableCheckpointing(5000);
        //将全局状态数据保存到 HDFS 或 内存中，rocksdb 上
        //flink1.13-版本
        //将全局的中间结果保存到 HDFS 上
        env.setStateBackend(new FsStateBackend("hdfs://node1:8020/flink-checkpints"));
        //存储 rocksdb
        env.setStateBackend(new RocksDBStateBackend("hdfs://node1:8020/flink-checkpints"));
        //flink1.13+版本
        //存储到 jobManager 内存
        env.setStateBackend(new HashMapStateBackend());
        //存储到 HDFS 上
        env.getCheckpointConfig().setCheckpointStorage("hdfs://node1:8020/flink-checkpoints/");
        //存储到 rocksdb 上，大于大状态、长窗口 rocksdb 上
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://node1:8020/flink-checkpoints/");
        //externalized checkpoint 当job被用户取消的时候是否将生成 checkpoint 都删除还是保留可以根据具体的业务规则
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //设置并行checkpoint
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //设置checkpoint timeout 超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000);
        //设置每个 checkpint 的间隔时间
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //设置 checkpoint 容忍失败的次数
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(10);
        //设置重启策略
        env.setRestartStrategy(RestartStrategies.noRestart());
        // 3）指定数据源
        env.socketTextStream("node1", 9999)
                // 4）对字符串进行空格拆分，然后每个单词记一次数
                //T value, Collector<O> out
                .flatMap((String line, Collector<Tuple2<String, Integer>> ctx) -> {
                    Arrays.stream(line.split(" ")).map(x -> Tuple2.of(x, 1)).forEach(ctx::collect);
                }).returns(Types.TUPLE(Types.STRING, Types.INT))
                // 5）对每个单词进行分组聚合操作
                .keyBy(t -> t.f0)
                .sum(1)
                // 6) 打印测试
                .print();
        // 7）执行任务，递交作业
        env.execute();
    }
}
